//package com.sure0000.basicKafka.example;
//
//import com.alibaba.fastjson.JSONArray;
//import com.alibaba.fastjson.JSONObject;
//import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
//import io.confluent.kafka.serializers.KafkaAvroSerializer;
//import org.apache.avro.Schema;
//import org.apache.avro.generic.GenericData;
//import org.apache.avro.generic.GenericRecord;
//import org.apache.kafka.clients.producer.KafkaProducer;
//import org.apache.kafka.clients.producer.Producer;
//import org.apache.kafka.clients.producer.ProducerConfig;
//import org.apache.kafka.clients.producer.ProducerRecord;
//
//import java.util.ArrayList;
//import java.util.List;
//import java.util.Properties;
//
///**
// * @program: demo
// * @description: producer 相关操作
// * @author: xuyouchang@supcon.com
// * @create: 2019-12-03 13:56
// **/
//public class KafkaProducerDemo {
//
//    public void avroProducer(String bootstrap, int num, String topic, String schemaUrl)  {
//        Schema.Parser parser=new Schema.Parser();
//        Schema schema=parser.parse(avroSchema());
//        Producer<String, GenericRecord> producer = new KafkaProducer<>(getProperties(bootstrap, schemaUrl));
//        List<ProducerRecord<String,GenericRecord>> records = getAvroRecords(schema, num, topic);
//        records.forEach((record) -> {
//            producer.send(record);
//        });
//    }
//
//    private List<ProducerRecord<String, GenericRecord>> getAvroRecords(Schema schema, int num, String topic) {
//        List<ProducerRecord<String, GenericRecord>> records = new ArrayList<>();
//        for (int i = 0; i < num; i++) {
//            GenericRecord avroRecord = new GenericData.Record(schema);
//            avroRecord.put("id",i);
//            avroRecord.put("name","jack4"+i);
//            avroRecord.put("age", 10+i);
//            ProducerRecord<String, GenericRecord> record = new ProducerRecord<>(topic, i+"jack", avroRecord);
//            records.add(record);
//        }
//        return records;
//    }
//
//    private Properties getProperties(String bootstrap, String schemaUrl) {
//        Properties props=new Properties();
//        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrap);
//        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
//        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
//        props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaUrl);
//
//        return props;
//    }
//
//    /** 为了应对 schema 变化，每个字段必须要有默认值 */
//    private String avroSchema() {
//        JSONObject schema = new JSONObject();
//        schema.put("type", "record");
//        schema.put("name", "Student");
//
//        JSONObject id = new JSONObject();
//        id.put("name", "id");
//        id.put("type", "int");
////        id.put("default", 0);
//
//        JSONObject name = new JSONObject();
//        name.put("name", "name");
//        name.put("type", "string");
////        name.put("default", "null");
//
//        JSONObject age = new JSONObject();
//        age.put("name", "age");
//        age.put("type", "int");
////        age.put("default", 0);
//
//        JSONArray fields = new JSONArray();
//        fields.add(id);
//        fields.add(name);
//        fields.add(age);
//        schema.put("fields", fields);
//        return schema.toJSONString();
//    }
//}
